package study.cdc.debezium.support.debezium;

import cn.hutool.core.thread.ThreadUtil;
import io.debezium.engine.DebeziumEngine;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.Assert;

import java.util.Random;

@Slf4j
public class DebeziumServer implements InitializingBean, SmartLifecycle {

    private volatile boolean running = false;

    private Thread t;

    String status = "ACTIVE";

    ZooKeeper zk;

    @Setter
    private DebeziumEngine<?> engine;

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(engine, "debezium engine must not be null");
        String connectString = "150.158.122.249:2181";
        zk = new ZooKeeper(connectString, 50000, null);
    }

    /**
     * （1）我们主要在该方法中启动任务或者其他异步服务，比如开启MQ接收消息
     * （2）当上下文被刷新（所有对象已被实例化和初始化之后）时，将调用该方法，默认生命周期处理器将检查每个SmartLifecycle对象的isAutoStartup()方法返回的布尔值。
     * 如果为“true”，则该方法会被调用，而不是等待显式调用自己的start()方法。
     */
    @Override
    public void start() {
        log.info(">>>>>>>>>>>>>>> start");
        try {
            String IP = "202.106.29.138";
            String port = new Random().nextInt(500) + "";
            if (zk.exists("/zk01", new MasterWatcher()) != null) { // 节点已经创建
                status = "STANDBY";
                log.info("等待standby");
                ThreadUtil.sleep(60 * 60 * 1000);
            } else {
                zk.create("/zk01", (IP + ":" + port).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                status = "ACTIVE";
                log.info("启动active");
                t = new Thread(engine, "debezium-engine-thread");
                t.start();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        //
        running = true;
    }

    public class MasterWatcher implements Watcher {

        public void process(WatchedEvent event) {
            log.info(">>>>>> WatcherWatcher");
            if (event.getType() == Event.EventType.NodeDeleted) {
                status = "ACTIVE";
                log.info(">>>>>> 切换为active");
                start();
            }
        }
    }


    @SneakyThrows
    @Override
    public void stop() {
        if (engine != null) {
            engine.close();
        }
        //
        running = false;
    }

    @Override
    public boolean isRunning() {
        return running;
    }
}
